package bsp.scan;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.sync.SyncException;

import util.ScanMessage;
import util.functors.OpScan;
import bsp.CMapper;

public class ScanStepOne<T extends Number> extends
		CMapper<LongWritable, Text, Text, NullWritable, ScanMessage> {

	private long chunkey = -1;
	private int myId;
	private String myName;
	private int numPeers;
	private OpScan<T> plus; // OpScan<T> : vb1

	public ScanStepOne(OpScan<T> op) {
		plus = op;
	}

	@Override
	public void map(LongWritable ki, Text line, Text ko, NullWritable vo,
			BSPPeer<LongWritable, Text, Text, NullWritable, ScanMessage> context)
			throws IOException, SyncException {
		//map
		T sum = plus.id();
		StringTokenizer itr = new StringTokenizer(line.toString().trim());
		while (itr.hasMoreTokens()) {
			sum = plus.evaluate(sum, str2Num(itr.nextToken()));
		}
		
		//send messages
		int peerId = myId + 1;
		while (peerId < numPeers) {
			addMesg(context.getPeerName(peerId), new ScanMessage(chunkey,
					sum.toString()));
			System.out.println(myName + " sends to "
					+ context.getPeerName(peerId));
			peerId++;
		}

	}

	@Override
	public void mapAll(
			BSPPeer<LongWritable, Text, Text, NullWritable, ScanMessage> context)
			throws IOException, SyncException, InterruptedException {

		LongWritable key = new LongWritable(-1);
		// only support long
		T sum = plus.id();
		Text line = new Text();
		// StringBuffer newList = new StringBuffer(512);
		while (context.readNext(key, line)) {
			if (chunkey < 0)
				chunkey = key.get();
			StringTokenizer itr = new StringTokenizer(line.toString().trim());
			while (itr.hasMoreTokens()) {
				sum = plus.evaluate(sum, str2Num(itr.nextToken()));
			}
		}
		// send p-sum to all
		// key: last line-key

		if (chunkey >= 0) {
			int peerId = myId + 1;
			while (peerId < numPeers) {
				addMesg(context.getPeerName(peerId), new ScanMessage(chunkey,
						sum.toString()));
				System.out.println(myName + " sends to "
						+ context.getPeerName(peerId));
				peerId++;
			}
		}
	}

	protected String Num2Str(T val) {
		return val.toString();
	}

	protected double str2Num(String token) {
		return Double.parseDouble(token);
	}

}